package com.larry.mapreduce.outputformat;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;
import java.security.Key;

/**
 * @ProjectName MapReduceDemo
 * @Author larry
 * @Date 20:35 2021/11/8
 * @Description TODO
 **/
public class LogRecordWriter extends RecordWriter<Text, NullWritable> {

    private FSDataOutputStream test1;
    private FSDataOutputStream test2;

    public LogRecordWriter(TaskAttemptContext job) throws IOException {
        FileSystem fs = FileSystem.get(job.getConfiguration());

        test1 = fs.create(new Path("C:\\Users\\larry\\IdeaProjects\\MapReduceDemo\\src\\main\\resources\\inputoutputformat\\output\\test1.log"));
        test2 = fs.create(new Path("C:\\Users\\larry\\IdeaProjects\\MapReduceDemo\\src\\main\\resources\\inputoutputformat\\output\\test2.log"));

    }

    @Override
    public void write(Text text, NullWritable value) throws IOException, InterruptedException {
        //具体写出
        String log = text.toString();
        if (log.contains("atguigu")){
            test1.writeBytes(log + "\n");
        }else {
            test2.writeBytes(log + "\n");
        }
    }

    @Override
    public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        IOUtils.closeStream(test1);
        IOUtils.closeStream(test2);
    }
}
